/**
 * 
 */
package org.metaverse.h2;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;

import javax.sql.DataSource;

import org.metaverse.h2.sql.SQL;
import org.metaverse.warehouse.BulkUploader;
import org.openrdf.model.BNode;
import org.openrdf.model.Literal;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.singularity.io.CloseablePipe;
import org.singularity.io.DME.Code;
import org.singularity.io.DME;
import org.singularity.io.Input;
import org.singularity.io.Output;

/**
 * @author vjache
 * 
 */
public class BulkUploaderImpl implements BulkUploader {

	private ValueFactory _valueFactory = new ValueFactoryImpl();
	
	private DataSource _datasource;
	private SQL _sql;

	/**
	 * @throws Exception
	 * 
	 */
	public BulkUploaderImpl(SQL aSql, DataSource aDataSource) throws Exception {
		_datasource = aDataSource;
		_sql = aSql;
		
//		new Thread(new Pump<Object[]>(null, null, "Bulk Uploader Pump")).start();
	}

	private void statementToBatch(Statement aStatement, DME.Code aCode,
			PreparedStatement aDmeInsert) throws SQLException {
		int pos = 1;
		
		//Subject #1,2
		final Resource subject = aStatement.getSubject();
		if (subject instanceof BNode) {
			aDmeInsert.setString(pos++, subject.stringValue());
			aDmeInsert.setString(pos++, "[");
		} else if (subject instanceof URI) {
			URI uri = (URI) subject;
			aDmeInsert.setString(pos++, uri.stringValue());
			aDmeInsert.setString(pos++, "<");
		} else
			throw new IllegalArgumentException(
					"Nither a URI nor Blank Node at subject.");

		//Predicate #3
		final URI predicate = aStatement.getPredicate();

		aDmeInsert.setString(pos++, predicate.stringValue());

		//Object #4,5
		final Value object = aStatement.getObject();
		if (object instanceof BNode) {
			aDmeInsert.setString(pos++, object.stringValue());
			aDmeInsert.setString(pos++, "[");
		} else if (object instanceof URI) {
			URI uri = (URI) object;
			aDmeInsert.setString(pos++, uri.stringValue());
			aDmeInsert.setString(pos++, "<");
		} else if (object instanceof Literal) {
			Literal lit = (Literal) object;
			aDmeInsert.setString(pos++, lit.stringValue());
			final URI datatype = lit.getDatatype();
			final String language = lit.getLanguage();

			if (datatype != null)
				aDmeInsert.setString(pos++, datatype.stringValue());
			else if (language != null && language.length() != 0)
				aDmeInsert.setString(pos++, "@" + language);
			else
				aDmeInsert.setNull(pos++, Types.VARCHAR);
		} else
			throw new IllegalArgumentException("Unreacheable code.");
		
		//Context #6
		final Resource context = aStatement.getContext();
		
		if (context instanceof URI) {
			URI uri = (URI) context;
			aDmeInsert.setString(pos++, uri.stringValue());
		}
		else if(context == null)
			aDmeInsert.setNull(pos++, Types.VARCHAR);
		else
			throw new IllegalArgumentException(
					"Only URIs allowed for contexts.");
		
		aDmeInsert.setString(pos++, "DefaultUser");

		aDmeInsert.setString(pos++, aCode == DME.Code.Add ? "+" : "-");

		aDmeInsert.addBatch();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.metaverse.warehouse.Uploader#getValueFactory()
	 */
	@Override
	public ValueFactory getValueFactory() {
		return _valueFactory;
	}

	@Override
	public Output<Statement> bulkModifyPush() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public CloseablePipe<Statement> bulkModifyPush(final Code aOpCode) {

		try {
			final Connection conn = _datasource.getConnection();
			final PreparedStatement dmeInsert = conn.prepareStatement(_sql
					.getDMEInsert());

			return new CloseablePipe<Statement>() {
				
				private int _cnt=0;

				@Override
				public void push(Statement aItem) throws IOException {
					try {
						if((_cnt%1000)==0)
						{
							flush();
							System.out.println(_cnt);
						}
						statementToBatch(aItem, aOpCode, dmeInsert);
						_cnt++;
					} catch (SQLException e) {
						throw new IOException(e);
					}
				}

				@Override
				public void flush() throws IOException {
					try {
						dmeInsert.executeBatch();
						dmeInsert.clearBatch();
					} catch (SQLException e) {
						throw new IOException(e);
					}
				}

				@Override
				public void close() throws IOException {
					flush();
					try {
						dmeInsert.close();
						conn.close();
					} catch (SQLException e) {
						throw new IOException(e);
					}
				}
				
				@Override
				protected void finalize() throws Throwable {
					close();
				}
			};
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void bulkModifyPull(Input<Statement> aInput) {
		// TODO Auto-generated method stub

	}

}
